package com.itcast.spark.test

import org.apache.spark.{SparkConf, SparkContext}
import org.apache.spark.streaming.{Seconds, StreamingContext}
import org.apache.spark.streaming.dstream.{DStream, ReceiverInputDStream}
import org.apache.spark.streaming.flume.{FlumeUtils, SparkFlumeEvent}

/**
 * DESC:
 */
object FlumeUtilOp {
  def main(args: Array[String]): Unit = {
    //1-准备环境
    val conf: SparkConf = new SparkConf().setAppName("SparkStreamingTCPTopK").setMaster("local[*]")
    val sc = new SparkContext(conf)
    sc.setLogLevel("WARN")
    //这里就是指定配置项将数据按照5秒为周期进行处理
    val ssc = new StreamingContext(sc, Seconds(5))
    //2-读取数据
    val data: ReceiverInputDStream[SparkFlumeEvent] = FlumeUtils.createPollingStream(ssc,"node01",8888)
    val receiveData: DStream[String] = data.map(x => new String(x.event.getBody.array()))
    //3-窗口的统计
    val result: DStream[(String, Int)] = receiveData.flatMap(_.split(" ")).map(x => (x, 1)).reduceByKeyAndWindow((x: Int, y: Int) => x + y, Seconds(10), Seconds(5))
    //4-结果输出print
    result.print()
    //5-ssc.start
    ssc.start()
    //6-ssc.awaitTermination
    ssc.awaitTermination()
    //7-ssc.stop(true,true)
    ssc.stop(true, true)
  }
}
